package com.kafka.utils;

import com.alibaba.fastjson.JSONObject;
import com.common.Common;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * @author xin.ding
 * @date 2019/12/7 14:24
 */
public class Producer {

    private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class);

    /**
     * 创建Kafka生产者
     *
     * @return KafkaProducer
     */
    private static KafkaProducer<String, String> createProducer() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", Common.KAFKA_BOOTSTRAP_SERVER);
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return new KafkaProducer<>((properties));
    }

    /**
     * 传入kafka约定的topicName,json格式字符串，发送给kafka集群
     *
     * @param topicName
     * @param jsonMessage
     */
    public static void sendMessage(String topicName, String jsonMessage) {
        KafkaProducer<String, String> producer = createProducer();
        producer.send(new ProducerRecord<>(topicName, jsonMessage));
        LOGGER.debug("meg:{}", jsonMessage);
        producer.close();
    }

    /**
     * 传入kafka约定的topicName,json格式字符串数组，发送给kafka集群<br>
     * 用于批量发送消息，性能较高。
     *
     * @param topicName
     * @param jsonMessages
     * @author IG
     * @Date 2017年4月14日 下午2:00:12
     * @version 1.0.0
     */
    public static void sendMessage(String topicName, String... jsonMessages) {
        KafkaProducer<String, String> producer = createProducer();
        for (String jsonMessage : jsonMessages) {
            producer.send(new ProducerRecord<>(topicName, jsonMessage));
            LOGGER.debug("meg:{}", jsonMessage);
        }
        producer.close();
    }

    /**
     * 传入kafka约定的topicName,Map集合，内部转为json发送给kafka集群 <br>
     * 用于批量发送消息，性能较高。
     *
     * @param topicName
     * @param mapMessageToJSONForArray
     * @author IG
     * @Date 2017年4月14日 下午2:01:18
     * @version 1.0.0
     */
    public static void sendMessage(String topicName, List<Map<Object, Object>> mapMessageToJSONForArray) {
        KafkaProducer<String, String> producer = createProducer();
        for (Map<Object, Object> map : mapMessageToJSONForArray) {
            String array = JSONObject.toJSONString(map);
            producer.send(new ProducerRecord<>(topicName, array));
        }
        producer.close();
    }




}
